package com.hlz.flink.chapter05;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.kafka.common.protocol.types.Field;

import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author Hongliang Zhu
 * @create 2023-01-02 20:47
 */
public class ClickSourceTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        System.out.println(env.getParallelism()); // 12
//        env.setParallelism(1);

//        DataStreamSource<Event> clickSource = env.addSource(new ClickSource()).setParallelism(1);

//        clickSource.print("my-source");

        DataStreamSource<Integer> integerDataStreamSource = env.addSource(new CustomSource()).setParallelism(4);
        integerDataStreamSource.print("integer-parallel");
        env.execute();
    }

    private static class CustomSource implements ParallelSourceFunction<Integer> {

        private boolean running = true;

        @Override
        public void run(SourceContext<Integer> ctx) throws Exception {

            Random random = new Random();

            while (running){
                int i = random.nextInt();
                ctx.collect(i);
                TimeUnit.SECONDS.sleep(1);
            }

        }

        @Override
        public void cancel() {

            running = false;
        }
    }

}
